package mapReduce;

import java.io.IOException;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;

import util.Utils;
import vo.Comment;
import vo.Photo;

public class RemoveCommentsWithoutPhotosMR {

	public static class Mapper extends
			TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {

		protected void map(ImmutableBytesWritable key, Result result,
				Context context) throws IOException, InterruptedException {

			String text = Bytes.toString(result
					.getRow());

			ImmutableBytesWritable immutableKeyPhoto = new ImmutableBytesWritable(
					Bytes.toBytes(text.substring(0, Utils.SIZE_UUID*2)));
			
			ImmutableBytesWritable immutableKeyComment = new ImmutableBytesWritable(Bytes.toBytes(text));
			context.write(immutableKeyPhoto, immutableKeyComment);
		}

	}

	public static class Reducer
			extends
			TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, NullWritable> {

		private static Integer quantidade = new Integer(0);

		protected void reduce(ImmutableBytesWritable key,
				Iterable<ImmutableBytesWritable> it, Context context)
				throws IOException, InterruptedException {

			HTable table = new HTable(Photo.TABLENAME);
			Get get = new Get(key.get());
			
			HTable tableComment = new HTable(Comment.TABLENAME);
			
			if(!table.exists(get)){
				for (ImmutableBytesWritable immutableBytesWritable : it) {
					tableComment.delete(new Delete(immutableBytesWritable.get()));
					quantidade++;
				}
			}
			
		}

		public static String printQteCommentRemoved() {
			
			return quantidade.toString();
		}

	}
}
